package com.zf.nospring.rabbitmq.consume_manual_ack_nack_reject;

import com.rabbitmq.client.*;
import com.zf.nospring.rabbitmq.utils.RabbitUtils;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * 首先要关闭自动确认
 * 手动拒绝消费，单条拒绝，以及被拒绝后是否重新入队
 * <p>
 * void basicReject(long deliveryTag, boolean requeue) 拒绝单条消息
 */
public class ConsumerReject {
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = RabbitUtils.getDefaultConnection();
        Channel channel1 = connection.createChannel();
        // 消费队列，关闭自动确认
        channel1.basicConsume("test_ack", false, getDefaultConsumer(channel1));
    }

    /**
     * 创建消费者的消费逻辑
     *
     * @param channel 通道
     */
    public static Consumer getDefaultConsumer(Channel channel) {
        // 创建消费者的消费逻辑
        return new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                // 消息的标识，基于通道，每个通道都是从1开始，当前通道每次读取一条消息值就+1, 用于消息的消费确认
                long deliveryTag = envelope.getDeliveryTag();
                // 获取当前通道
                Channel ch = getChannel();

                // 如果消息是重新入队的，则随机拒绝或者确认
                if (envelope.isRedeliver()) {
                    if (deliveryTag % 2 == 0) {
                        ch.basicReject(deliveryTag, false);
                    } else {
                        ch.basicAck(deliveryTag, false);
                    }
                    System.out.printf("requeue: channel: %s\n consumerTag：%s，deliveryTag: %d，msg: %s\n", ch.toString(), consumerTag, deliveryTag, new String(body));
                    return;
                }

                System.out.printf("channel: %s\n consumerTag：%s，deliveryTag: %d，msg: %s\n", ch.toString(), consumerTag, deliveryTag, new String(body));
                // 拒绝消费单条消息，并重新入队
                /*void basicReject(long deliveryTag, boolean requeue) 拒绝消费单条消息
                     deliveryTag：用来标识消息的id
                     requeue：被拒绝的消息是否重新入队，如果使用不当容易造成 消费->被拒绝->重新入队->消费->被拒绝... 的死循环中
                 */
                ch.basicReject(deliveryTag, true);
            }
        };
    }
}
